/********************************************************/
/*	Copyright (C) 2012 Gong Li Bin			*/
/*	Project:	MapReduce			*/
/*	Author:		gong_libin			*/
/*	Date:		2016_06_01			*/
/*	File:		MrAccept.cpp			*/
/********************************************************/

#include "MrAccept.h"

CMrUdp g_CUdp;
CMrSet g_CPoll;
CMrSet g_CProc;
MRJBPACKET_S g_stProc;
bool g_bDebug = false;

int main(int argc, char* argv[])
{
	int iCount = 0;
	int iIndex = 0;
	int iReturn = 0;
	int iThread = 1;
	CMrThread CAdmin;
	CMrThread* pCThread = NULL;
	char szPort[MR_VALUE] = { 0 };
	char szConf[MR_VALUE] = "MrAccept.conf";

	struct option stOptions[] = {
		{"conf", 1, 0, 'c'},
		{"thread", 1, 0, 't'},
		{"debug", 0, 0, 'd'},
		{"help", 0, 0, 'h'},
		{"version", 0, 0, 'v'},
		{0, 0, 0, 0},
	};

	while (true) {
		iReturn = getopt_long(argc, argv, "c:t:dhv", stOptions, &iIndex);
		if (iReturn < 0) {
			break;
		}
		switch (iReturn) {
			case 'c':
				strncpy(szConf, optarg, MR_VALUE);
				break;
			case 't':
				iThread = atoi(optarg);
				break;
			case 'd':
				g_bDebug = true;
				break;
			case 'h':
				MrJbHelp(argv[0], MR_AC_VERSION);
				return MR_SUCCESS;
			case 'v':
				MR_PRINT("Copyright (C) 2012 Gong Li Bin Technology Inc.\n");
				MR_PRINT("%s Build on %s %s.\n", MR_AC_VERSION, __DATE__, __TIME__);
				return MR_SUCCESS;
			default:
				return MR_SUCCESS;
		}
	}

	signal(SIGSEGV, MrJbQuit);
	signal(SIGTERM, MrJbQuit);

	/* config */
	if (true == g_bDebug) {
		MR_DEBUG("Initialize config from %s\n", szConf);
	}
	if (MR_SUCCESS == MrAcceptConfig(szConf, szPort)) {
		if (true == g_bDebug) {
			MR_DEBUG("Port: %s\n", szPort);
		}
	}
	else {
		return MR_FAILURE;
	}

	/* fork */
	if (true != g_bDebug) {
		MrFork();
	}

	/* admin */
	if (true == g_bDebug) {
		MR_DEBUG("Create admin thread\n");
	}
	CAdmin.MrThreadAttrInit();
	CAdmin.MrThreadSetStartRoutine(MrJbAdminCore);
	if (MR_FAILURE == CAdmin.MrThreadCreate(NULL)) {
		return MR_FAILURE;
	}

	/* accept */
	if (true == g_bDebug) {
		MR_DEBUG("MrAccept will listen...\n");
	}
	if (MR_SUCCESS == g_CUdp.MrNetInitRecv(NULL, szPort)) {
		if (true == g_bDebug) {
			iThread = 1;
		}
		pCThread = new CMrThread[iThread];
		for (iCount = 0; iCount < iThread; iCount ++) {
			(pCThread + iCount)->MrThreadAttrInit();
			(pCThread + iCount)->MrThreadSetStartRoutine(MrAcceptThreadCore);
			(pCThread + iCount)->MrThreadCreate(NULL);
		}
		for (iCount = 0; iCount < iThread; iCount ++) {
			(pCThread + iCount)->MrThreadJoin(NULL);
		}
		g_CUdp.MrNetClose();
	}
	else {
		MR_ERROR("Failed to MrNetInitRecv\n");
	}

	return MR_SUCCESS;
}

int MrAcceptConfig(char* pszConf, char* pszPort)
{
	CMrDb CDb;
	MRCONFIG_S stConf;
	USHORT usCount = 0;
	int iReturn = MR_SUCCESS;

	memset(&stConf, '\0', sizeof(MRCONFIG_S));
	if (MR_SUCCESS == MrConfig(pszConf, &stConf)) {
		if (MR_SUCCESS == CDb.MrDbConnect(stConf.m_szHost, stConf.m_szPort)) {
			/* port */
			iReturn = MrJbPortInit(CDb, MR_AC_PORT, &stConf, pszPort);
			/* poll */
			if (MR_SUCCESS == iReturn) {
				if (MR_SUCCESS == (iReturn = MrJbDevInit(CDb, MR_AC_POLL, &stConf, &g_CPoll, MR_JB_POLL))) {
					if (true == g_bDebug) {
						MR_DEBUG("Poll:");
						for (usCount = 0; usCount < g_CPoll.MrSetGetCount(); usCount ++) {
							MR_PRINT(" %d", ((MRJBPOLL_S*)g_CPoll.MrSetIndexGet(usCount))->m_stHost.m_uiID);
						}
						MR_PRINT("\n");
					}
				}
			}
			/* proc */
			if (MR_SUCCESS == iReturn) {
				if (MR_SUCCESS == (iReturn = MrJbDevInit(CDb, MR_AC_PROC, &stConf, &g_CProc, MR_JB_HOST))) {
					if (true == g_bDebug) {
						MR_DEBUG("Proc:");
						for (usCount = 0; usCount < g_CProc.MrSetGetCount(); usCount ++) {
							MR_PRINT(" %d", ((MRJBHOST_S*)g_CProc.MrSetIndexGet(usCount))->m_uiID);
						}
						MR_PRINT("\n");
					}
					MrJbDevReqSet(&g_stProc, g_CProc);
				}
			}
			CDb.MrDbClose();
		}
		else {
			MR_ERROR("Failed to MrDbConnect\n");
			iReturn = MR_FAILURE;
		}
	}
	else {
		MR_ERROR("Failed to MrConfig\n");
		iReturn = MR_FAILURE;
	}

	return iReturn;
}

void* MrAcceptThreadCore(void* pNULL)
{
	CMrPro CPro;
	CMrUdp CSend;
	CMrUdp CRecv;
	CMrTime CTime;
	int iLength = 0;
	char szUrl[MR_URL] = { 0 };
	UCHAR ucResult = MR_DEF_SUCC;
	UINT uiProc[MR_JB_DEVICE] = { 0 };
	struct sockaddr_in* pstAdp = NULL;
	UCHAR uszSend[MR_BIGPACKET] = { 0 };
	UCHAR uszRecv[MR_BIGPACKET] = { 0 };
	MRPROKV_S stKv[MR_JB_PAIR] = {{MR_MR_ADP, 0, (UCHAR*)szUrl}, {'\0', 0, NULL}};

	srand((UINT)time(NULL));
	CRecv.MrNetSetSocket(g_CUdp.MrNetGetSocket());
	if (MR_SUCCESS == CSend.MrNetOpen(MR_SOCK_DGRAM)) {
		CSend.MrNetSetTimeOut(0, MR_USEC);
		while ((iLength = CRecv.MrNetRecv(uszRecv, MR_BIGPACKET)) > 0) {
			if (true == g_bDebug) {
				MR_PRINT("\n");
				CTime.MrTimeOn();
				MR_DEBUG("Thread[%lu] start\n", pthread_self());
			}

			/* head */
			CPro.MrProReset();
			CPro.MrProParse(uszRecv, iLength);
			if (true == g_bDebug) {
				MR_DEBUG("<Request>:\n");
				CPro.MrProDisplay();
			}

			/* real */
			if (MR_PRO_0 != CPro.MrProGetReal()) {
				MrJbResponse((struct sockaddr_in*)CRecv.MrUdpGetAddr());
			}

			/* body */
			if (CPro.MrProGetLength() > 0) {
				if (true == g_bDebug) {
					MR_DEBUG("MrProKvInput:\t(Recv -> KV)\n");
				}
				memset(stKv + 1, '\0', sizeof(MRPROKV_S) * (MR_JB_PAIR - 1));
				MrProKvInput(stKv + 1, CPro.MrProGetTotal(), CPro.MrProGetBody(), CPro.MrProGetLength());

				if (true == g_bDebug) {
					MR_DEBUG("MrProKvOutput:\t(KV -> Send)\n");
				}
				memset(szUrl, '\0', MR_URL);
				memset(uszSend, '\0', MR_BIGPACKET);
				pstAdp = (struct sockaddr_in*)CRecv.MrUdpGetAddr();
				sprintf(szUrl, "%s:%d", inet_ntoa(pstAdp->sin_addr), ntohs(pstAdp->sin_port));
				if (true == g_bDebug) {
					MR_DEBUG("MrAdapter: %s\n", szUrl);
				}
				stKv[0].m_iLength = strlen(szUrl);
				CPro.MrProPutTotal(CPro.MrProGetTotal() + 1);
				CPro.MrProPutAttr(CPro.MrProGetAttr() | MR_MR_ACC);
				CPro.MrProPutLength(CPro.MrProGetLength() + MR_PRO_KEYLEN + stKv[0].m_iLength);
				MrProKvOutput(stKv, CPro.MrProGetTotal(), &uszSend[MR_PRO_HEAD], MR_BIGPACKET - MR_PRO_HEAD);
			}
			else {
				ucResult = MR_DEF_LENG;
			}

			/* poll */
			if (MR_DEF_SUCC == ucResult) {
				memset(uiProc, 0, MR_JB_DEVICE);
				ucResult = MrJbDevSort(CSend, &g_stProc, uiProc);
				if (true == g_bDebug) {
					MR_DEBUG("MrJbDevSort: 0x%02x, ", ucResult);
				}
				switch (ucResult) {
					case MR_DEF_SUCC:
						break;
					case MR_DEF_NULL:
						break;
					case MR_DEF_SEND:
						MrJbDevInput(uiProc, g_CProc);
						ucResult = MR_DEF_SUCC;
						break;
					case MR_DEF_RECV:
						MrJbDevInput(uiProc, g_CProc);
						ucResult = MR_DEF_SUCC;
						break;
					case MR_DEF_UNKN:
						MrJbDevInput(uiProc, g_CProc);
						ucResult = MR_DEF_SUCC;
						break;
					default:
						break;
				}
				if (true == g_bDebug) {
					MrJbDevDisplay(uiProc);
				}
			}

			/* packet */
			if (true == g_bDebug) {
				MR_DEBUG("MrProCreate: ");
			}
			if (MR_DEF_SUCC != ucResult) {
				CPro.MrProPutTotal(MR_PRO_0);
				CPro.MrProPutLength(MR_PRO_0);
			}
			CPro.MrProPutBody(NULL);
			CPro.MrProPutResult(ucResult);
			CPro.MrProCreate(uszSend, MR_PRO_HEAD);

			/* send */
			if (MR_DEF_SUCC != ucResult) {
				if (true == g_bDebug) {
					MR_PRINT("0x%02x, Will sendto MrAdapter\n", ucResult);
					MR_DEBUG("<MrAdapter>:\n");
					CPro.MrProDisplay();
				}
				CRecv.MrNetSend(uszSend, MR_PRO_HEAD);
			}
			else {
				if (true == g_bDebug) {
					MR_PRINT("0x%02x, Will sendto MrProcess\n", ucResult);
					MR_DEBUG("<MrProcess>:\n");
					CPro.MrProDisplay();
				}
				MrAcceptProcExec(CSend, uiProc, uszSend, MR_PRO_HEAD + CPro.MrProGetLength());
			}

			if (true == g_bDebug) {
				MR_DEBUG("Thread[%lu] end Use time ", pthread_self());
				CTime.MrTimeOff();
				CTime.MrTimeDisplay();
			}

			memset(uszRecv, '\0', MR_BIGPACKET);
			ucResult = MR_DEF_SUCC;
		}
		CSend.MrNetClose();
	}

	pthread_exit(NULL);
}

void MrAcceptProcExec(CMrUdp& rCSend, UINT* puiProc, UCHAR* puszSend, int iLength)
{
	CMrPro CPro;
	int iRecv = 0;
	USHORT usProc = 0;
	MRJBHOST_S stHost;
	UCHAR uszHead[MR_PRO_HEAD] = { 0 };

	while (0 != *(puiProc + usProc)) {
		memset(&stHost, '\0', sizeof(MRJBHOST_S));
		if (MR_SUCCESS == g_CProc.MrSetSearch(puiProc + usProc, &stHost)) {
			rCSend.MrUdpSetAddr(&stHost.m_unAddr.m_stAddr);
			if (rCSend.MrNetSend(puszSend, iLength) > 0) {
				if (true == g_bDebug) {
					MR_DEBUG("Send message to proc %d\n", *(puiProc + usProc));
				}
				memset(uszHead, '\0', MR_PRO_HEAD);
				if ((iRecv = rCSend.MrNetRecv(uszHead, MR_PRO_HEAD)) > 0) {
					CPro.MrProParse(uszHead, iRecv);
					if (MR_DEF_SUCC == CPro.MrProGetResult()) {
						if (true == g_bDebug) {
							MR_DEBUG("Result: success from proc %d\n", *(puiProc + usProc));
						}
						break;
					}
					else {
						if (true == g_bDebug) {
							MR_DEBUG("Result: failure from proc %d\n", *(puiProc + usProc));
						}
					}
				}
				else {
					if(true == g_bDebug) {
						MR_DEBUG("Failed to MrNetRecv\n");
					}
				}
			}
			else {
				if (true == g_bDebug) {
					MR_DEBUG("Failed to MrNetSend\n");
				}
			}
		}
		else {
			if (true == g_bDebug) {
				MR_DEBUG("Failed to MrSetSearch\n");
			}
		}
		usProc += 1;
	}

	return;
}
